package dht.threads;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.LinkedList;

import dht.communication.CommunicationDHT;
import dht.communication.PortThread;
import dht.data.DhtData;
import dht.data.Options;
import dht.data.TabBigInteger;
import dht.interfaceThread.InterGet;
import dht.message.GetMessage;
import dht.message.Message;
import dht.node.Node;
import dht.tools.OperationsZone;

public class ManageGet extends Thread implements InterGet {
	
	/**
	 * Tableau des jobs que l'on peut donner au ManageGet
	 */
	private LinkedList<TabBigInteger> jobs=new LinkedList<TabBigInteger>();
	
	/**
	 * Socket de communication du trhead
	 */
	private DatagramSocket maSocket;

	/**
	 * Booléen indiquant si le noeud est pret ou non
	 */
	public Boolean isBusy=new Boolean(Boolean.TRUE);
	
	/**
	 * Port de la socket
	 */
	public static long portSocket;
	
	/**
	 * Fonction de réception d'une requête GET
	 */
	public void receiveRequestGet() {
		GetMessage m=null;
		try {
			m=(GetMessage) CommunicationDHT.receiveMsg(maSocket);
			Options.update_socketGetTimeout(false, true);
			maSocket.setSoTimeout(Options.socket_get_timeout);
		} catch (SocketTimeoutException e) {
			Options.update_socketGetTimeout(true, false);
			try {
				maSocket.setSoTimeout(Options.socket_get_timeout);
			} catch (SocketException e1) {
				e1.printStackTrace();
			}
		} catch (IOException e) {
		} catch (ClassNotFoundException e) {
		}
		if (m==null) {
			return;
		}
		if (m.TYPE_MSG==Message.REPLY_Message) {
			Node.putGetReply(m.getReply());
			return;
		}
		if (m.TYPE_MSG==Message.MISS_Message) {
			Node.putGetReply(null);
			return ; 
		}
		boolean b=false;
		synchronized(Node.myContext){
			b=OperationsZone.isInZone(m.getDestination(), Node.myContext.getZoneLocale());
		}
		if (b) {
			ackClient(m,GetData(m)); //méthode d'acquittement du client
		}
		else {
			int index=OperationsZone.IsInContexteNonFusion(m.getDestination());
			
			if (index!=-1) {
				DhtData<Integer> d=null;
				synchronized (Node.contexteNonFusion.get(index)) {
					d=Node.contexteNonFusion.get(index).getDhtLocale().get(m.getDestination());
				}
				ackClient(m,d); //méthode d'acquittement du client
			}
			else {
				GetMessage msg = new GetMessage(m.getadresseIpExp(),m.getVirtualId(),m.getDestination());
				this.sendRequestGet(msg);
			}
		}
	
	}

	/**
	 * Fonction de retour d'une valeur
	 * @param m le message reçu à traiter
	 * @return la valeur sotckée dans la table de hachage locale
	 */
	private DhtData<Integer> GetData(GetMessage m) {
		synchronized (Node.myContext) {
			return Node.myContext.getDhtLocale().get(m.getDestination());
		}
	}

	/**
	 * Fonction d'acquittement du client ayant fait le Get
	 * @param p le message reçu à traiter
	 */
	private void ackClient(GetMessage p,DhtData<Integer> data) {
		InetAddress inet=null;
		try {
			inet = InetAddress.getByName(CommunicationDHT.getLocalIp());
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
		GetMessage msg = new GetMessage(inet,"",p.getDestination());
		
		msg.setReply(data);
		
		if (data!=null) msg.TYPE_MSG=Message.REPLY_Message;
		else msg.TYPE_MSG=Message.REPLY_Message;

		DatagramPacket pp=CommunicationDHT.packetForgeToKnownNode(msg, p.getadresseIpExp(), PortThread.PORT_GET);
		CommunicationDHT.sendMessage(pp, maSocket);
	}



	/**
	 * Fonction appellée par le noeud pour donner un job au manageGET
	 * @param job le job à donner au manageGET
	 */
	public void putJob(TabBigInteger job) {
		synchronized (jobs) {
			jobs.offer(job);
		}
	}

	public void run() {
		try {
			maSocket=new DatagramSocket(PortThread.PORT_GET);
			maSocket.setSoTimeout(Options.socket_get_timeout);  // set the socket timeout to 20 ms
		} catch (SocketException e1) {
			e1.printStackTrace();
		}

		while (true) {
			
			boolean b=true;
			while (b) {
				synchronized(isBusy) {
					if (isBusy.booleanValue()!=true) {
						b=false;
					}
				}
				try {
					if (b!=false) sleep(10);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			TabBigInteger toTreat=null;
			synchronized(jobs) {
				if (jobs.size()!=0) {
					toTreat=jobs.poll();
				}
			}
			if (toTreat!=null) {
				boolean truc=false;
				synchronized (Node.myContext) {
					if (OperationsZone.isInZone(toTreat, Node.myContext.getZoneLocale())) {
						truc=true;
					}
				}
				if (truc == true) {
					synchronized (Node.myContext) {
						Node.putGetReply(Node.myContext.getDhtLocale().get(toTreat));
					}
				}
				else {
					// ici , on doit envoyer un get sur le réseau
					InetAddress monIP=null;
					try {
						monIP=InetAddress.getByName(CommunicationDHT.getLocalIp());;
					} catch (UnknownHostException e) {
						e.printStackTrace();
					}
	
					GetMessage msg = null;
					
					synchronized(Node.myContext){
						msg = new GetMessage(monIP,Node.myContext.getVirtualID(),toTreat);
					}
	
					this.sendRequestGet(msg);
				}
			}

			this.receiveRequestGet();
		}
	}

	@Override
	public boolean sendRequestGet(GetMessage g) {
		DatagramPacket p= CommunicationDHT.packetForgeToClosestOrRandomNode(g, PortThread.PORT_GET);
		return CommunicationDHT.sendMessage(p, maSocket);
	}
}
